package distributed;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

import common.*;

public class Middleware extends Thread{
	public RoutingTable<NodeRecord, Node> routing_table;
	private ArrayList<Integer> ip_pool;
	private Long system_up_time;
	long ip = 3232235521L;	// always 192.168.0.1
	public ConcurrentLinkedQueue<Message> msg_q;	// a message box for middleware(public)
	
	private ArrayList<Node> reported_dead_nodes;
	
	private ArrayList<Node> nodes_tested;
	private ArrayList<Integer> time_lapse;
	
	private int msg_id = 0;
	
	public Middleware(){
		routing_table = new RoutingTable<NodeRecord, Node>();
		ip_pool = new ArrayList<Integer>();
		
		msg_q = new ConcurrentLinkedQueue<Message>();
		
		reported_dead_nodes = new ArrayList<Node>();
		nodes_tested = new ArrayList<Node>();
		time_lapse = new ArrayList<Integer>();
		
		// generates available IP pool (prevent duplicate IP's are generated by rand()).
		for( int i = 0; i < Config.MAX_NODES; i++ ){
			while( true ){
				Integer gen = ( new Random() ).nextInt( Config.MAX_NODES * 2 ) + 2;
				
				if( !ip_pool.contains( gen ) ){
					ip_pool.add( gen );
					break;
				}
			}
		}
		
		// initial NTP time receiving.
		try{
			system_up_time = Utils.get_ntp_time();
		}catch( IOException e ){
			e.printStackTrace();
		}//System.currentTimeMillis();
	}
	
	public void run(){	// periodically prints the current time
		while( true ){
			Message incoming_msg = msg_q.poll();
			while(incoming_msg != null)
			{
				switch( incoming_msg.getHeader()){
				case Def.REQ:
					if(incoming_msg.getContent().substring(0,4).equals("dead"))
					{
						String s1 = incoming_msg.getContent().substring(4);
						StringTokenizer st1 = new StringTokenizer(s1, ";");
						while(st1.hasMoreTokens())
						{
							String s2 = st1.nextToken();
							StringTokenizer st2 = new StringTokenizer(s2,",");
							NodeRecord node_record = new NodeRecord(Integer.parseInt(st2.nextToken()),Long.parseLong(st2.nextToken()));
							Node node = routing_table.get(node_record);
							if(node!=null)
							{
								if(!reported_dead_nodes.contains(node)&&!nodes_tested.contains(node))
								{
									reported_dead_nodes.add(node);
								}
							}
						}
					}
				case Def.ACK:
					if(incoming_msg.getContent().equals("yes"))
					{
						NodeRecord record = new NodeRecord(incoming_msg.getSpid(), incoming_msg.getSip());
						Node node = routing_table.get(record);
						if(nodes_tested.contains(node))
						{
							time_lapse.remove(nodes_tested.indexOf(node));
							nodes_tested.remove(node);
							send(incoming_msg.getMid(), Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "OK");
						}
						else{
							print_log("Error: received msg from dead nodes", Def.STDERR);
						}
					}
				}
				incoming_msg = msg_q.poll();
			}
			
			if(!reported_dead_nodes.isEmpty())
			{
				while(!reported_dead_nodes.isEmpty())
				{
					Node node = reported_dead_nodes.get(0);
					nodes_tested.add(node);
					time_lapse.add(0);
					send(++msg_id, Def.MW_REQ, node.get_pid(), node.get_ip(), "alive?");  
				    reported_dead_nodes.remove(0);
				}
			}
			
			try {
				sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			for(int i = 0; i < nodes_tested.size(); i++)
			{
				int time = time_lapse.get(i) + 1;
				time_lapse.remove(i);
				time_lapse.add(i,time);
			}
			
			for(int i = 0; i < nodes_tested.size(); i++)
			{
				int time = time_lapse.get(i);
				Node node = nodes_tested.get(i);
				// first time out
				if(time == Config.REQUEST_TIMEOUT_S)
				{
					send(++msg_id, Def.MW_REQ, node.get_pid(), node.get_ip(), "alive?");
				}
				// second time out
				if(time == 2*Config.REQUEST_TIMEOUT_S)
				{
					send(++msg_id, Def.MW_REQ, node.get_pid(), node.get_ip(), "alive?");
				}
				// third time out, decide that the node is dead
				if(time == 3*Config.REQUEST_TIMEOUT_S)
				{
					NodeRecord record = new NodeRecord(node.get_pid(),node.get_ip());
					routing_table.remove(record);
					
					String notification = "DROP:" + Integer.toString(node.get_pid()) + "," + Long.toString(node.get_ip());
					send(++msg_id, Def.MW_NOTIFY, 0, 0L, notification);
				}
			}
			
//			print_log( "Current time = " + System.currentTimeMillis() + " (ms)" );
			
//			try{
//				sleep( 1000 * 60 * 10 );	// 10 min
//			}catch( InterruptedException e ){
//				e.printStackTrace();
//			}
		}
	}
	
	/**
	 * Method: connect() - allows nodes to connect to the network
	 * @return newIp = ip addr assigned to the node
	 * 				or -1 on err
	 */
	public synchronized long connect( int new_pid, Node new_node ){	// --- should be in CS
		long new_ip = -1;
		
		if( ip_pool.isEmpty() ){
			print_log( "no IP is available. The network is full.", Def.STDERR );
			return new_ip;
		}
		
		new_ip = ip_pool.remove( 0 );
		new_ip += 3232235520L;	// 192.168.0.x
		
		routing_table.put( new NodeRecord( new_pid, new_ip ), new_node ); //newMacAddr ) );
		String notification = "Add :" + Integer.toString(new_pid) + "," + Long.toString(new_ip); 
		send(++msg_id, Def.MW_NOTIFY, 0, 0L, notification);
				
		// only the first node gets NTP time as the initial value
		try{
			new_node.set_localtime_s( Utils.get_ntp_time() );
		}catch ( IOException e ){
			e.printStackTrace();
		}
		//new_node.set_localtime_ms( System.currentTimeMillis() );
		
		return new_ip;
	}
	
	public synchronized boolean disconnect( long ip, int pid ){	// --- should be in CS
		int old_ip = ( int )( ip - 3232235520L );	// 192.168.0.x
		
		if( ip_pool.contains( old_ip ) ){
			print_log( "IP " + ip + " has been already released.", Def.STDERR );
			return false;
		}
		
		if( ip_pool.add( old_ip ) ){
			routing_table.remove( new NodeRecord( pid, ip ) );
			print_log( "IP " + ip + " has been successfully released." );
			return true;
		}else
			print_log( "IP " + ip + " cannot be released.", Def.STDERR );
		
		return false;
	}
	
	public int send( int msg_id, int msg_header, int rcv_pid, Long rcv_ip, String appx)
	{
		//print_log( "**middleware msg** " + msg_header );
		
		Message outgoing_msg = new Message( msg_id, msg_header, rcv_pid, rcv_ip, -1, -1L );
		if( appx != null ){
			outgoing_msg.setContent( appx );
		}
		
		// send msg
		if( rcv_pid == 0 ){	// 0 to broadcast
			Collection<Node> all_nodes = routing_table.values();
			Iterator<Node> itr = all_nodes.iterator();
			int br_count = 0;	// #msg broadcasted
			while( itr.hasNext() ){
				Node br_rcv = itr.next();	// br_rcv = broadcast, receiver
				
			    br_rcv.msg_q.offer( outgoing_msg );
			    br_count++;
			}
			//print_log( "br_msg(" + msg_header + ") is sent to " + br_count + " nodes in the network (" + outgoing_msg.getMid() + ")." );		
			
			return br_count;
			
		}
		else{
				Node receiver = routing_table.get( new NodeRecord( rcv_pid, rcv_ip ) );
				receiver.msg_q.offer( outgoing_msg );
			}			
			return 1;
	}
	
	public void print_log( String log ){
		print_log( log, Def.STDOUT );
	}
	
	public void print_log( String log, byte target ){
		if( !Config.VERBOSE )
			return;
		
		long ntp = 0;
		
		try{
			ntp = Utils.get_ntp_time();
		}catch( IOException e ){
			e.printStackTrace();
		}
		
		
		if( target == Def.STDOUT ){
			//System.out.print( "*[MSG|router:" + Utils.ipToString( ip ) + "(" + System.currentTimeMillis() + ")] " );
			System.out.print( "*[MSG|router:" + Utils.ipToString( ip ) + "(" + ntp + ")] " );
			System.out.println( log );
		}else if( target == Def.STDERR ){
			//System.err.print( "*[ERR|router:" + Utils.ipToString( ip ) + "(" + System.currentTimeMillis() + ")] " );
			System.err.print( "*[ERR|router:" + Utils.ipToString( ip ) + "(" + ntp + ")] " );
			System.err.println( log );
		}
	}

	public class RoutingTable<K, V> extends Hashtable{
		public synchronized V get( Object key, long ip1, long ip2 ){
			int n_hops;
			int delay;
			
			// calc # hops
			if( ip1 / Config.SUBNET_SIZE == ip2 / Config.SUBNET_SIZE ){
				n_hops = 0;
			}else{
				n_hops = 4;
			}
			
			// sleep( delay )
			// delay is hop delay * gaussian factor (normal btwn .5 and 1.5)
			delay = ( int )( n_hops * Config.DELAY_HOP * ( ( new Random() ).nextGaussian() + .5 ) );
			
			try{
				sleep( delay );
			}catch( InterruptedException e ){ e.printStackTrace(); }

			// return V;
			return ( V )super.get( key );
		}
		
		public synchronized V get(Object key)
		{
			return (V)super.get(key);
		}
	}
}
